// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: donggu

#pragma once

#include <map>
#include <memory>
#include <string>
#include <utility>
#include <vector>

#include "coin2/exchange/base/executor/topic_publisher.h"
#include "coin2/exchange/base/feed/live_writer/processor.h"
#include "coin2/exchange/base/feed/processor.h"
#include "coin2/exchange/base/feed/system.h"
#include "coin2/flow/topic_subscription.h"
#include "coin2/exchange/base/feed/ws/subscriber_manager.h"
#include "coin2/exchange/base/market/market_exchange_api.h"
#include "coin2/exchange/factory/symbology/product.h"
#include "coin2/flow/topic_string.h"

namespace coin2::exchange::base::feed {
namespace impl {

using coin2::base::config::ValidateWebsocketConfig;
using coin2::exchange::feed::FeedSystemConfig;

class LiveFeedWriterSubsystem : public BaseLiveFeedSubsystem {
 public:
  LiveFeedWriterSubsystem(
      const std::string& system_name,
      const MarketExchangeApi& mea,
      const di::ConstantInjection& ci,
      const FeedSystemConfig& config,
      api_base::LiveAsyncExecutor* live_executor)
      : BaseLiveFeedSubsystem(system_name, mea, live_executor),
        processor_(mea, config.archive(), live_executor_->WriterIoc()) {
    subscriber_ = std::make_shared<SubscriberManager>(
        ci.GetFeedSubscriptionSpec(mea),
        config.websocket(),
        coin2::flow::TopicType::LONG_TOPIC,
        live_executor,
        config.raw_parser().enable_telemetry());

    CHECK_THROW(!config.products().resolve_realtime())
        << "(resolve_realtime=true) is not supported yet";

    products_ = ProductRequestsFromConfig(mea_.String(), config.products());
    for (auto& prod : products_) {
      auto symbol = prod->feed_native_symbol();
      char group = flow::SymbolGroupConfig::GetInstance().GetGroupForRelativeNorm(
          mea_,
          prod->relative_norm());
      subscriber_->RequestNativeSymbol(symbol, group);
    }
  }

  void StartAsync() override {
    LOG(INFO) << "[LiveFeedWriter] StartAsync() RegisterLiveLongTopic";
    topic_register_.RegisterLiveLongTopic(mea_, &processor_, live_executor_->distributor());
    subscriber_->RegisterPubTopic();
    subscriber_->StartAsync();
    health_.status = HealthStatus::HEALTH_STARTING;
    health_.since = GetCurrentTimestamp();
  }
  void StopAsync() override {
    subscriber_->StopAsync();
  }

  std::vector<IProduct*> GetProducts() const override {
    std::vector<IProduct*> out;
    for (const auto& p : products_) {
      out.push_back(p.get());
    }
    return out;
  }

  void GetHealthReport(int indent, std::string* report) const override {
    // [HealthReport]
    //     [Feed]
    //         - Huobi (OK)
    //         - Okex (STARTING, uptime=23h)
    //               a1 (..)
    HealthReport health{};
    subscriber_->GetHealthReport(indent + 1, &health);

    report->append(std::string(indent * 4, ' '));
    std::string info = fmt::format(
        "- {system} ({health}, uptime='{uptime}', connection={num})\n",
        fmt::arg("system", system_name()),
        fmt::arg("health", HealthStatusName(this->health_.status)),
        fmt::arg("uptime", DebugStringFromDuration(GetCurrentTimestamp() - health_.since)),
        fmt::arg("num", health.connection_num));
    report->append(info).append(health.report);
  }

 private:
  std::shared_ptr<SubscriberManager> subscriber_;
  live_writer::FeedWriterProcessor processor_;
  std::vector<std::unique_ptr<IProduct>> products_;
  TopicRegister topic_register_;
};

}  // namespace impl

using impl::LiveFeedWriterSubsystem;
}  // namespace coin2::exchange::base::feed
